package edu.nepu.bigdata.utils;

import cn.hutool.core.io.FileUtil;
import com.alibaba.fastjson.JSONObject;
import edu.nepu.bigdata.constant.Constant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class CommandLineParser {
    private static final Map<String, String> arguments = new HashMap<>();
    private static Logger logger = LoggerFactory.getLogger(CommandLineParser.class);
    // kafka连接参数
    private static final ArrayList<String> requiredKafkaArgs = new ArrayList<String>(){{
        add("topic");
        add("bootstrap.servers");
        add("fullClassName");
        add("methodName");
    }};

    // redis连接参数
    private static final ArrayList<String> requiredRedisArgs = new ArrayList<String>(){{
        add("connectType");
        add("hostPort");
        //add("pikaMaster");
        add("password");
        add("fullClassName");
        add("methodName");
    }};

    /**
     * 解析参数
     * @param args 命令行参数
     */
    private static void parseParam(String[] args) throws Exception {
        for (String param : args) {
            if (param.startsWith("--")) {
                String[] keyAndVal = param.split("=");
                if (keyAndVal.length != 2) {
                    throw new IllegalArgumentException("Invalid argument format: " + param);
                }
                String key = keyAndVal[0].substring(2);
                String value = keyAndVal[1];
                arguments.put(key, value);
            }
        }
        // 加载外部连接
        String url = "./connection_info.txt";
        List<String> kafkaConnect = FileUtil.readLines(url, StandardCharsets.UTF_8);
        HashMap<String, String> connectMap = new HashMap<>();
        for (String connectInfo : kafkaConnect) {
            try {
                JSONObject jsonObject = JSONObject.parseObject(connectInfo);
                String env = jsonObject.getString("env");
                String bootstrapServers = jsonObject.getString("bootstrap.servers");
                connectMap.put(env,bootstrapServers);
            } catch (Exception e){
                throw new Exception("无法解析连接信息，请确保传入的参数符合json格式: " + connectInfo);
            }
        }
        String kafkaAddress = arguments.get("bootstrap.servers");
        arguments.put("bootstrap.servers", connectMap.getOrDefault(kafkaAddress, kafkaAddress));
    }

    /**
     * 校验连接kafka的参数
     * @param args 命令行参数
     */
    public static void parseKafkaArgs(String[] args) throws Exception {
        parseParam(args);
        // 检查参数是否符合标准
        for (String requiredArg : requiredKafkaArgs) {
            if (!arguments.containsKey(requiredArg)) {
                throw new IllegalArgumentException("Missing required argument: " + requiredArg);
            }
        }
        if (CommandLineParser.getArg("timestamp") == null && CommandLineParser.getArg("date_hms") == null){
            throw new IllegalArgumentException("Missing required argument: timestamp or date_hms");
        }
        logger.info("传递的参数：");
        arguments.forEach((k, v) -> logger.info("Argument: " + k + "=" + v));
    }

    /**
     * 校验连接redis的参数
     * @param args 命令行参数
     */
    public static void parseRedisArgs(String[] args) throws Exception {
        parseParam(args);
        // 检查参数是否符合标准
        for (String requiredArg : requiredRedisArgs) {
            if (!arguments.containsKey(requiredArg)) {
                throw new IllegalArgumentException("Missing required argument: " + requiredArg);
            }
        }
        if (arguments.get("connectType").equalsIgnoreCase(Constant.REDIS_SENTINEL) || arguments.get("connectType").equalsIgnoreCase("s")){
            if (arguments.get("pikaMaster") == null){
                throw new IllegalArgumentException("Missing required argument: pikaMaster");
            }
        }
        logger.info("传递的参数：");
        arguments.forEach((k, v) -> logger.info("Argument: " + k + "=" + v));
    }

    public static String getArg(String key) {
        return arguments.get(key);
    }
    public static boolean containsArg(String key) {
        return arguments.containsKey(key);
    }
}

